package raft

//
// this is an outline of the API that raft must expose to
// the service (or tester). see comments below for
// each of these functions for more details.
//
// rf = Make(...)
//   create a new Raft server.
// rf.Start(command interface{}) (index, term, isleader)
//   start agreement on a new log entry
// rf.GetState() (term, isLeader)
//   ask a Raft for its current term, and whether it thinks it is leader
// ApplyMsg
//   each time a new entry is committed to the log, each Raft peer
//   should send an ApplyMsg to the service (or tester)
//   in the same server.
//

import (
	`bytes`
	`log`
	"sync"
	`sync/atomic`

	"../labgob"
	`../labrpc`
)

// 所有可能的操作
//
//
// as each Raft peer becomes aware that successive log entries are
// committed, the peer should send an ApplyMsg to the service (or
// tester) on the same server, via the applyCh passed to Make(). set
// CommandValid to true to indicate that the ApplyMsg contains a newly
// committed log entry.
//
// in Lab 3 you'll want to send other kinds of messages (e.g.,
// snapshots) on the applyCh; at that point you can add fields to
// ApplyMsg, but set CommandValid to false for these other uses.
//
type ApplyMsg struct {
	CommandValid bool
	Command      interface{}
	CommandIndex int
}

// 和选举密切相关的状态
type raftTermState struct {
	// paper request
	currentTeam int
	voteFor     int

	// raft request
	identity uint8
}

// 和日志拷贝密切相关的状态
type raftLogState struct {
	mu          sync.Mutex
	logs        []logEntry
	commitIndex int
	lastApplied int
}

// Leader特有的状态
type raftLeaderState struct {
	mu              sync.Mutex
	nextIndex       []int
	isSynchronizing []bool
}

type Raft struct {
	// raw
	mu        sync.Mutex          // Lock to protect shared access to this peer's state
	peers     []*labrpc.ClientEnd // RPC end points of all peers
	persister *Persister          // Object to hold this peer's persisted state
	me        int                 // this peer's index into peers[]
	dead      int32               // set by Kill()

	// paper request
	HeartBeatHelper
	raftTermState
	raftLogState
	raftLeaderState

	// lab request
	applyCh chan ApplyMsg // applyCh is a channel on which the tester or service expects Raft to send ApplyMsg messages.
}

type raftPersistState struct {
	CurrentTerm int
	VoteFor     int
	Logs        []logEntry
}

func (rf *Raft) isLeader() bool {
	return rf.identity == Leader
}

// return currentTerm and whether this server
// believes it is the leader.
func (rf *Raft) GetState() (int, bool) {
	return rf.currentTeam, rf.isLeader()
}

func (rf *Raft) GetPeers() []*labrpc.ClientEnd {
	line := make([]*labrpc.ClientEnd, len(rf.peers))
	copy(line, rf.peers)
	return line
}

//
// save Raft's persistent state to stable storage,
// where it can later be retrieved after a crash and restart.
// see paper's Figure 2 for a description of what should be persistent.
//
func (rf *Raft) persist(lockLog bool) {
	// Your code here (2C).
	if lockLog {
		rf.raftLogState.mu.Lock()
		defer rf.raftLogState.mu.Unlock()
	}
	w := new(bytes.Buffer)
	rfState := raftPersistState{Logs: rf.logs, CurrentTerm: rf.currentTeam, VoteFor: rf.voteFor}
	encoder := labgob.NewEncoder(w)
	e := encoder.Encode(rfState)
	//w, e := json.Marshal(rfState)
	if e != nil {
		log.Fatalln("Marshal raftPersisState fail, detail: ", e)
	}
	rf.persister.SaveRaftState(w.Bytes())
}

//
// restore previously persisted state.
//
func (rf *Raft) readPersist(data []byte) {
	if data == nil || len(data) < 1 { // bootstrap without any state?
		return
	}
	rfState := new(raftPersistState)

	r := bytes.NewBuffer(data)
	decoder := labgob.NewDecoder(r)
	e := decoder.Decode(&rfState)

	//e := json.Unmarshal(data, &rfState)
	if e != nil {
		log.Fatalln("Unmarshal raftPersisState fail, detail: ", e)
	}
	rf.logs = rfState.Logs
	rf.currentTeam = rfState.CurrentTerm
	rf.voteFor = rfState.VoteFor
}

//
// the service using Raft (e.g. a k/v server) wants to start
// agreement on the next command to be appended to Raft's log. if this
// server isn't the leader, returns false. otherwise start the
// agreement and return immediately. there is no guarantee that this
// command will ever be committed to the Raft log, since the leader
// may fail or lose an election. even if the Raft instance has been killed,
// this function should return gracefully.
//
// the first return value is the index that the command will appear at
// if it's ever committed. the second return value is the current
// term. the third return value is true if this server believes it is
// the leader.
//
func (rf *Raft) Start(command interface{}) (int, int, bool) {
	rf.mu.Lock()
	defer rf.mu.Unlock()
	if !rf.isLeader() || rf.dead == 1 {
		DPrintf("[%d] get command, not leader, drop it", rf.me)
		return rf.commitIndex, rf.currentTeam, false
	}

	DPrintf("[%d] get command: %d, is leader, append log", rf.me, command)
	// return index, term, isLeader
	return rf.makeAppendLog(command), rf.currentTeam, true
}

//
// the tester doesn't halt goroutines created by Raft after each test,
// but it does call the Kill() method. your code can use killed() to
// check whether Kill() has been called. the use of atomic avoids the
// need for a lock.
//
// the issue is that long-running goroutines use memory and may chew
// up CPU time, perhaps causing later tests to fail and generating
// confusing debug output. any goroutine with a long-running loop
// should call killed() to check whether it should stop.
//
func (rf *Raft) Kill() {
	atomic.StoreInt32(&rf.dead, 1)
	// Your code here, if desired.
}

func (rf *Raft) killed() bool {
	z := atomic.LoadInt32(&rf.dead)
	return z == 1
}

func (rf *Raft) initLeader() {
	rf.raftLeaderState.mu.Lock()
	defer rf.raftLeaderState.mu.Unlock()
	index := rf.nowLogIndex() + 1
	rf.nextIndex = make([]int, len(rf.peers))
	for i := range rf.nextIndex {
		rf.nextIndex[i] = index
	}
	rf.isSynchronizing = make([]bool, len(rf.peers))
}

//
// the service or tester wants to create a Raft server. the ports
// of all the Raft servers (including this one) are in peers[]. this
// server's port is peers[me]. all the servers' peers[] arrays
// have the same order. persister is a place for this server to
// save its persistent state, and also initially holds the most
// recent saved state, if any. applyCh is a channel on which the
// tester or service expects Raft to send ApplyMsg messages.
// Make() must return quickly, so it should start goroutines
// for any long-running work.
//
func Make(peers []*labrpc.ClientEnd, me int, persister *Persister, applyCh chan ApplyMsg) *Raft {
	rf := &Raft{peers: peers, persister: persister, me: me, applyCh: applyCh}

	// initialize from state persisted before a crash
	rf.readPersist(persister.ReadRaftState())
	rf.becomeFollower()
	rf.updateTimeOut()
	go rf.alwaysCommitLogs()
	go rf.waitHeartBeatTimeOut()
	go rf.alwaysBroadcastEntries()
	return rf
}
